package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo3BatchSQL {
  def main(args: Array[String]): Unit = {
    /**
     * flink 流处理模式和批处理模式的的区别
     * 流处理模式
     * 1、流处理模式食醋胡连续结果
     * 2、上游task和下游task同时启动一起运行，等待数据到达
     * 批处理模式
     * 1、批处理模式输出最终结果
     * 2、批处理模式先执行上游task，再执行下游task，任务调度的方式和spark一致
     * 3、可以在上游做预聚合，提高计算的效率
     *
     */
    //环境设置对象
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      //指定处理模式
      //.inStreamingMode() //流处理模式
      .inBatchMode() //批处理模式,输出最终结果
      .build()

    //flink sql的执行环境
    val tEnv: TableEnvironment = TableEnvironment.create(settings)


    /**
     * 1、创建source表 -- file source表
     *
     */

    tEnv.executeSql(
      """
        |CREATE TABLE student (
        |  id STRING,
        |  name STRING,
        |  age INT,
        |  sex STRING,
        |  clazz STRING
        |) WITH (
        |  'connector' = 'filesystem',           -- 必选：指定连接器类型，filesystem，文件系统
        |  'path' = 'flink/data/students.txt',  -- 必选：指定路径
        |  'format' = 'csv',                     -- 必选：文件系统连接器指定 format
        |  'csv.field-delimiter' = ','
        |)
        |""".stripMargin)

    /**
     * 创建一个用于保存结果的表  -- file sink 表
     */

    tEnv.executeSql(
      """
        |CREATE TABLE clazz_num (
        | clazz STRING,
        | num BIGINT
        |) WITH (
        |  'connector' = 'filesystem',           -- 必选：指定连接器类型，filesystem，文件系统
        |  'path' = 'flink/data/clazz_num',  -- 必选：指定路径
        |  'format' = 'csv',                     -- 必选：文件系统连接器指定 format
        |  'csv.field-delimiter' = ','
        |)
        |
        |""".stripMargin)

    /**
     * 2、统计班级的人数
     *
     */

    tEnv.executeSql(
      """
        |insert into clazz_num
        |select clazz,count(1) as num
        |from student
        |group by clazz
        |
        |""".stripMargin)
  }

}
